package com.mask.im.sdk.service.impl;

import com.alibaba.fastjson.JSON;
import com.mask.im.sdk.dto.IMPushMessage;
import com.mask.im.sdk.service.MessagePushService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.Future;

/**
 * 基于Kafka的消息推送服务实现
 * 负责将消息发送到Kafka对应分区，实现跨节点消息传递
 *
 * @author jian
 */
@Slf4j
@Service
public class MessageKafkaPushServiceImpl implements MessagePushService {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    // Kafka生产者配置
    @Resource
    private KafkaProducer<String, String> kafkaProducer;

    @Value("${im.kafka.bootstrap-servers:localhost:9092}")
    private String bootstrapServers;

    @Value("${im.kafka.topic:im-messages}")
    private String kafkaTopic;

    @Value("${im.kafka.partitions:1}")
    private int kafkaPartitions;

    // Redis Key前缀 - 保留用户节点映射
    private static final String USER_NODE_PREFIX = "im:user:node:";
    private static final String GROUP_MEMBERS_PREFIX = "im:group:members:";

    /**
     * 销毁Kafka生产者
     */
    @PreDestroy
    public void destroy() {
        if (kafkaProducer != null) {
            try {
                kafkaProducer.close(java.time.Duration.ofSeconds(5));
                log.info("Kafka生产者已关闭");
            } catch (Exception e) {
                log.error("关闭Kafka生产者时发生异常", e);
            }
        }
    }

    @Override
    public <T> Boolean pushMessage(IMPushMessage<T> message) {
        if (message == null || message.getReceiverId() == null) {
            throw new IllegalArgumentException("消息或接收者ID不能为null");
        }

        if ("FRIEND".equals(message.getPushType())) {
            // 好友消息 - 单用户
            String nodeId = getUserNode(message.getReceiverId());

            if (nodeId == null) {
                log.warn("用户 {} 不在线，无法发送消息", message.getReceiverId());
                return null;
            }

            try {
                // 验证消息内容
                String messageJson = JSON.toJSONString(message);
                if (messageJson.trim().isEmpty()) {
                    log.error("消息JSON序列化结果为空，无法发送");
                    return null;
                }
                int partition = Math.abs(nodeId.hashCode()) % kafkaPartitions;
                ProducerRecord<String, String> record = new ProducerRecord<>(
                        kafkaTopic,
                        partition,
                        nodeId,
                        messageJson
                );
                kafkaProducer.send(record).get(5, java.util.concurrent.TimeUnit.SECONDS);
                log.info("好友消息已发送到Kafka，接收者: {}, 节点: {}",
                        message.getReceiverId(), nodeId);
                return true;
            } catch (Exception e) {
                log.error("发送好友消息失败", e);
                return false;
            }
        }
        return false;
    }

    @Override
    public <T> List<Long> pushGroupMessage(IMPushMessage<T> message) {
        if (message == null || message.getReceiverId() == null) {
            throw new IllegalArgumentException("消息或接收者ID不能为null");
        }
        if ("GROUP".equals(message.getPushType())) {
            // 按节点分组成员
            Map<String, List<Long>> nodeMembersMap = new HashMap<>();
            List<Long> offlineMembers = new ArrayList<>();
            List<Long> receiversId = message.getReceiversId();
            for (Long memberId : receiversId) {
                String nodeId = getUserNode(memberId);
                if (nodeId != null) {
                    nodeMembersMap.computeIfAbsent(nodeId, k -> new ArrayList<>()).add(memberId);
                } else {
                    offlineMembers.add(memberId);
                }
            }

            // 发送到各个节点
            boolean allSuccess = true;
            for (Map.Entry<String, List<Long>> entry : nodeMembersMap.entrySet()) {
                String nodeId = entry.getKey();
                List<Long> members = entry.getValue();
                // 为每个节点创建消息副本，包含该节点需要推送的成员列表
                IMPushMessage nodeMsg = createGroupMessageForNode(message, message.getGroupId(), members);
                String result = sendToGroupNode(nodeMsg, nodeId);
                if (result == null) {
                    offlineMembers.add(nodeMsg.getReceiverId());
                    allSuccess = false;
                }
            }
            log.info("群组消息已发送到 {} 个节点，离线成员 {} 个",
                    nodeMembersMap.size(), offlineMembers.size());
            return offlineMembers;
        }
        return new ArrayList<>();
    }

    @Override
    public Boolean isOnline(Long userId) {
        String userNode = getUserNode(userId);
        return StringUtils.isNotBlank(userNode);
    }

    @Override
    public Boolean isOnline(Long userId, String device) {
        String userNode = getUserNode(userId, device);
        return StringUtils.isNotBlank(userNode);
    }

    @Override
    public List<Long> getAllOnlineUser() {
        String pattern = USER_NODE_PREFIX + ":*";
        Set<String> keys = redisTemplate.keys(pattern);
        List<Long> onlineUserList = new ArrayList<>();
        if (keys != null && !keys.isEmpty()) {
            for (String key : keys) {
                // 使用冒号分割字符串
                String[] parts = key.split(":");
                // 目标值在分割后的数组索引为3（从0开始计数）
                if (parts.length >= 4) {
                    String target = parts[3];
                    onlineUserList.add(Long.valueOf(target));
                } else {
                    log.info("获取所有在线用户字符串格式不符合预期");
                }
            }
        }
        return onlineUserList;
    }


    /**
     * 注册用户到节点
     */
    public void registerUser(String userId, String device, String nodeId) {
        String key = USER_NODE_PREFIX + userId + ":" + device;
        redisTemplate.opsForValue().set(key, nodeId);
        log.info("用户 {} 设备 {} 已注册到节点 {}", userId, device, nodeId);
    }

    /**
     * 注销用户
     */
    public void unregisterUser(String userId, String device) {
        String key = USER_NODE_PREFIX + userId + ":" + device;
        redisTemplate.delete(key);
        log.info("用户 {} 设备 {} 已注销", userId, device);
    }

    /**
     * 根据用户ID查找节点（默认设备）
     */
    private String getUserNode(Long userId) {
        // 先尝试查找默认设备
        String key = USER_NODE_PREFIX + userId + ":default";
        Object nodeId = redisTemplate.opsForValue().get(key);
        if (nodeId != null) {
            return nodeId.toString();
        }

        // 如果没找到，查找所有设备
        String pattern = USER_NODE_PREFIX + userId + ":*";
        Set<String> keys = redisTemplate.keys(pattern);
        if (keys != null && !keys.isEmpty()) {
            Object firstNodeId = redisTemplate.opsForValue().get(keys.iterator().next());
            if (firstNodeId != null) {
                return firstNodeId.toString();
            }
        }
        return null;
    }

    /**
     * 根据用户ID和设备查找节点
     */
    public String getUserNode(Long userId, String device) {
        String key = USER_NODE_PREFIX + userId + ":" + device;
        Object nodeId = redisTemplate.opsForValue().get(key);
        return nodeId != null ? nodeId.toString() : null;
    }



    /**
     * 直接发送消息到指定节点（群组消息）
     */
    public String sendToGroupNode(IMPushMessage msg, String nodeId) {
        if (msg == null || nodeId == null) {
            throw new IllegalArgumentException("消息或节点ID不能为null");
        }

        try {
            // 验证消息内容
            String messageJson = JSON.toJSONString(msg);
            if (messageJson.trim().isEmpty()) {
                log.error("群组消息JSON序列化结果为空，无法发送");
                return null;
            }

            int partition = Math.abs(nodeId.hashCode()) % kafkaPartitions;
            ProducerRecord<String, String> record = new ProducerRecord<>(
                    kafkaTopic,
                    partition,
                    nodeId,
                    messageJson
            );

            kafkaProducer.send(record).get(5, java.util.concurrent.TimeUnit.SECONDS);
            log.info("群组消息已发送到节点 {}，发送者: {}", nodeId, msg.getSenderId());
            return "0";
        } catch (Exception e) {
            log.error("发送群组消息到指定节点失败", e);
            return "1";
        }
    }

    /**
     * 获取所有在线节点（通过Redis中的用户节点映射推断）
     */
    public List<String> getOnlineNodes() {
        String pattern = USER_NODE_PREFIX + "*";
        Set<String> keys = redisTemplate.keys(pattern);
        if (keys == null || keys.isEmpty()) {
            return new ArrayList<>();
        }

        Set<String> nodeIds = new HashSet<>();
        for (String key : keys) {
            Object nodeId = redisTemplate.opsForValue().get(key);
            if (nodeId != null) {
                nodeIds.add(nodeId.toString());
            }
        }

        return new ArrayList<>(nodeIds);
    }

    /**
     * 发送群组消息
     */
    public boolean sendGroupMessage(IMPushMessage msg, Long groupId) {
        if (msg == null || groupId == null) {
            throw new IllegalArgumentException("消息或群组ID不能为null");
        }

        // 获取群组成员
        Set<Long> groupMembers = getGroupMembers(groupId);

        if (groupMembers == null || groupMembers.isEmpty()) {
            log.warn("群组 {} 没有成员，无法发送消息", groupId);
            return false;
        }

        // 按节点分组成员
        Map<String, List<Long>> nodeMembersMap = new HashMap<>();
        List<Long> offlineMembers = new ArrayList<>();

        for (Long memberId : groupMembers) {
            String nodeId = getUserNode(memberId);
            if (nodeId != null) {
                nodeMembersMap.computeIfAbsent(nodeId, k -> new ArrayList<>()).add(memberId);
            } else {
                offlineMembers.add(memberId);
            }
        }

        // 发送到各个节点
        boolean allSuccess = true;
        for (Map.Entry<String, List<Long>> entry : nodeMembersMap.entrySet()) {
            String nodeId = entry.getKey();
            List<Long> members = entry.getValue();
            // 为每个节点创建消息副本，包含该节点需要推送的成员列表
            IMPushMessage nodeMsg = createGroupMessageForNode(msg, groupId, members);
            String result = sendToGroupNode(nodeMsg, nodeId);
            if (result == null) {
                allSuccess = false;
            }
        }

        log.info("群组消息已发送到 {} 个节点，离线成员 {} 个",
                nodeMembersMap.size(), offlineMembers.size());
        return allSuccess;
    }

    /**
     * 为节点创建群组消息副本
     */
    private <T> IMPushMessage<T> createGroupMessageForNode(IMPushMessage<T> pushMessage, Long groupId, List<Long> members) {
        IMPushMessage<T> message = new IMPushMessage<T>();
        message.setSenderId(pushMessage.getSenderId());
        message.setReceiverId(pushMessage.getReceiverId());
        message.setGroupId(groupId);
        message.setReceiversId(members);
        message.setContent(pushMessage.getContent());
        message.setPushType("GROUP");
        return message;
    }

    /**
     * 添加群组成员
     */
    public void addGroupMember(String groupId, String userId) {
        String key = GROUP_MEMBERS_PREFIX + groupId;
        redisTemplate.opsForSet().add(key, userId);
        log.info("用户 {} 已加入群组 {}", userId, groupId);
    }

    /**
     * 移除群组成员
     */
    public void removeGroupMember(String groupId, String userId) {
        String key = GROUP_MEMBERS_PREFIX + groupId;
        redisTemplate.opsForSet().remove(key, userId);
        log.info("用户 {} 已退出群组 {}", userId, groupId);
    }

    /**
     * 批量添加群组成员
     */
    public void addGroupMembers(String groupId, List<String> userIds) {
        if (userIds == null || userIds.isEmpty()) {
            return;
        }

        String key = GROUP_MEMBERS_PREFIX + groupId;
        redisTemplate.opsForSet().add(key, userIds.toArray());
        log.info("已批量添加 {} 个用户到群组 {}", userIds.size(), groupId);
    }

    /**
     * 获取群组成员
     */
    public Set<Long> getGroupMembers(Long groupId) {
        String key = GROUP_MEMBERS_PREFIX + groupId;
        Set<Object> members = redisTemplate.opsForSet().members(key);
        if (members == null) {
            return new HashSet<>();
        }

        Set<Long> stringMembers = new HashSet<>();
        for (Object member : members) {
            if (member != null) {
                stringMembers.add(Long.valueOf(member.toString()));
            }
        }
        return stringMembers;
    }

    /**
     * 获取群组成员数量
     */
    public Long getGroupMemberCount(String groupId) {
        String key = GROUP_MEMBERS_PREFIX + groupId;
        return redisTemplate.opsForSet().size(key);
    }

    /**
     * 检查用户是否在群组中
     */
    public boolean isGroupMember(String groupId, String userId) {
        String key = GROUP_MEMBERS_PREFIX + groupId;
        return Boolean.TRUE.equals(redisTemplate.opsForSet().isMember(key, userId));
    }

    /**
     * 删除群组
     */
    public void deleteGroup(String groupId) {
        String key = GROUP_MEMBERS_PREFIX + groupId;
        redisTemplate.delete(key);
        log.info("群组 {} 已删除", groupId);
    }
}
